package com.minelx.datamicroscope.core.condition;

import com.alibaba.fastjson.JSONObject;
import com.minelx.datamicroscope.common.Configurations;
import com.minelx.datamicroscope.common.EnrichedIterator;
import com.minelx.datamicroscope.core.values.CommonField;
import io.gitee.minelx.commontools.factor.CloseableIterator;
import io.gitee.minelx.commontools.factor.Factor;

import java.util.function.Supplier;

public class SingleKeyStreamSourceFactorFactory implements ConditionFactorFactory {
    private static final Configurations keyDescriptions = Configurations.of("/key_descriptions.json");

    private static final int DEFAULT_PARTITION_TOLERANCE = 50;

    private final CommonField singleKey;

    private final Supplier<CloseableIterator<String>> iteratorSupplier;

    public SingleKeyStreamSourceFactorFactory(CommonField singleKey, Supplier<CloseableIterator<String>> iteratorSupplier) {
        this.singleKey = singleKey;
        this.iteratorSupplier = iteratorSupplier;
    }

    @Override
    public Factor<Condition> factor() {
        return () -> EnrichedIterator.fromCloseableIterator(iteratorSupplier.get())
                // FIXME wait a moment...    IN EXPRESSION all the time even if partition counts is 1
                .pack(getPartitionTolerance(singleKey))
                .map(values -> KeyOperandCondition.of(singleKey, Operand.in(values)));
    }

    private static int getPartitionTolerance(CommonField key) {
        JSONObject config = (JSONObject) keyDescriptions.toJSONObject().getOrDefault(key.path(), defaultConfig());
        return config.getIntValue("partitionTolerance");
    }

    private static JSONObject defaultConfig() {
        JSONObject result = new JSONObject();
        result.put("partitionTolerance", DEFAULT_PARTITION_TOLERANCE);
        return result;
    }
}
